package demo.delay;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Configuration;
import org.springframework.util.CollectionUtils;

import java.util.HashMap;
import java.util.Map;

@Configuration
@ConfigurationProperties("spring.cloud.stream")
public class KafkaDelayConfig {

    private Map<String, String> bindings;

    private Map<String, String> kafka;

    /**
     * 获取消费端配置
     *
     * @return
     */
    public Map<String, Object> getConsumerProperties(KafkaDelayListener kafkaDelayListener) {

        Map<String, Object> configMap = new HashMap<>();

        if (CollectionUtils.isEmpty(bindings) || CollectionUtils.isEmpty(kafka)) {
            return configMap;
        }

        configMap.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafka.get("binder.brokers"));
        configMap.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_uncommitted");
        configMap.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        configMap.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        configMap.put(ConsumerConfig.GROUP_ID_CONFIG, bindings.get(kafkaDelayListener.channelName() + ".group"));
        configMap.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, Boolean.FALSE);

        return configMap;
    }

    public Map<String, String> getBindings() {
        return bindings;
    }

    public void setBindings(Map<String, String> bindings) {
        this.bindings = bindings;
    }

    public Map<String, String> getKafka() {
        return kafka;
    }

    public void setKafka(Map<String, String> kafka) {
        this.kafka = kafka;
    }

}
